package org.example;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * Flink 及时流处理（时间序列分析）示例
 * 本示例展示如何使用 Flink 进行时间序列分析：
 * 1. 从文本流中读取数据，每条数据包含时间戳。
 * 2. 使用 WatermarkStrategy 设置 Event Time 和 Watermark 生成策略。
 * 3. 按单词分组，使用 TumblingEventTimeWindows 进行窗口聚合，统计每 5 秒内单词出现次数。
 * 适合初学者学习 Flink 时间语义和窗口计算的基本使用。
 */
public class TimeSeriesAnalysisExample {

    public static void main(String[] args) throws Exception {
        // 创建流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为 1，方便观察输出
        env.setParallelism(1);

        // 模拟输入数据流，每条数据包含时间戳（实际应用中可能来自 Kafka、Socket 等）
        DataStream<String> textStream = env.fromElements(
            "Hello Flink, 1625097600000", // 2021-07-01 00:00:00
            "Hello World, 1625097605000", // 2021-07-01 00:00:05
            "Hello Flink, 1625097610000"  // 2021-07-01 00:00:10
        );

        // 解析输入数据，转换为 (word, 1) 的元组，同时提取时间戳
        DataStream<Tuple2<String, Integer>> wordCountStream = textStream
            .map(new MapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public Tuple2<String, Integer> map(String value) {
                    String[] parts = value.split(", ");
                    String text = parts[0];
                    long timestamp = Long.parseLong(parts[1]);
                    return new Tuple2<>(text, 1);
                }
            })
            // 设置 Event Time 和 Watermark 生成策略
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<Tuple2<String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                    .withTimestampAssigner((event, timestamp) -> Long.parseLong(event.f0.split(", ")[1]))
            )
            // 按单词分组，使用 TumblingEventTimeWindows 进行窗口聚合，统计每 5 秒内单词出现次数
            .keyBy(0)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .sum(1);

        // 输出结果
        wordCountStream.print("时间序列分析结果");

        // 执行流处理作业
        env.execute("Time Series Analysis Example");
    }
} 